package com.winit.schedule.job.tracker.event.persist;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

import com.winit.schedule.job.tracker.conf.JobConfiguration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.winit.schedule.core.event.EventHandler;
import com.winit.schedule.core.repository.entity.TaskRunRecordEntity;
import com.winit.schedule.job.tracker.JobTrackerContext;
import com.winit.schedule.job.tracker.repository.ScheduleDataManager;

public class RunRecordHandler implements EventHandler<DataPersistEvent> {

    private static final Logger                logger            = LoggerFactory.getLogger(TaskTrackerHandler.class);

    private ScheduleDataManager                scheduleDataManager;

    private JobTrackerContext                  jobTrackerContext;

    private List<Thread>                       dataHandlerThread = new ArrayList<Thread>();

    private BlockingQueue<TaskRunRecordEntity> dataQueue         = new LinkedBlockingQueue<TaskRunRecordEntity>();

    public RunRecordHandler(JobTrackerContext jobTrackerContext){
        this.scheduleDataManager = jobTrackerContext.getScheduleDataManager();
        this.jobTrackerContext = jobTrackerContext;

        for (int i = 1; i <= 3; i++) {
            Thread thread = new DataHandler(i);
            dataHandlerThread.add(thread);
            thread.start();
        }
    }

    @Override
    public void handle(DataPersistEvent event) {
        boolean developMode = jobTrackerContext.getConf().getBoolean(JobConfiguration.DEVELOP_MODE,
                JobConfiguration.DEFAULT_DEVELOP_MODE);

        if (!developMode && null != event && DataPersistEventType.LOG_RUN_RECORED.equals(event.getType())
            && null != event.getTaskRunRecord()) {
            boolean isSuccess = dataQueue.offer(event.getTaskRunRecord());
            if (!isSuccess) {
                logger.warn(">>>>>>>>>> Record run log fails, the queue is full <<<<<<<<<<");
            }
        }
    }

    public ScheduleDataManager getScheduleDataManager() {
        return scheduleDataManager;
    }

    public JobTrackerContext getJobTrackerContext() {
        return jobTrackerContext;
    }

    class DataHandler extends Thread {

        public DataHandler(int num){
            this.setName("Thread-TaskRunRecord-Persist-" + num);
        }

        @SuppressWarnings("static-access")
        @Override
        public void run() {
            while (true) {
                try {
                    TaskRunRecordEntity record = dataQueue.take();
                    if (null != record) {
                        String taskRunningUUID = String.valueOf(record.getTaskRunnerUuid());
                        synchronized (taskRunningUUID) {
                            scheduleDataManager.saveTaskRunRecord(record);
                        }
                    }

                    Thread.currentThread().sleep(50);
                } catch (Exception e) {
                    logger.error("TaskRunRecord persist failure.", e);
                }
            }
        }
    }
}
